package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.trip.CarTripOutputPhev;
import cn.gwm.flink.streaming.beans.trip.CarTripPhev;
import cn.gwm.flink.streaming.function.map.TripPhevMapFunction;
import cn.gwm.flink.streaming.function.sink.SinkHbasePhev;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import java.util.List;
import java.util.Properties;

public class TripDwmPhevTask extends BaseTask{
    public static void main(String[] args) throws Exception {
        //StreamExecutionEnvironment env = getEnv("trip");
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3,
                Time.seconds(30)
        ));
        env.setParallelism(3);
/*        final FlinkKafkaConsumer<String> kafkaConsumer = FlinkKafkaUtil.getConsumerGroupOffsets(
                "consumer-dwd-clean-phev",
                ConfigLoader.get("kafka.clean.topic.dwd.phev", "dwd_clean_phev")

        );*/
        Properties prop = new Properties();
        prop.put("bootstrap.servers", ConfigLoader.get("bootstrap.servers"));
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer(
                //Kafka主题
                ConfigLoader.get("kafka.clean.topic.dwd.phev", "dwd_clean_phev"),
                //"dwd_clean_phev_trip",
                //序列化方式
                new SimpleStringSchema(),
                //Kafka连接地址
                prop);
        SingleOutputStreamOperator<String> kafkaOperator = env.addSource(kafkaConsumer.setStartFromLatest());
        //SingleOutputStreamOperator<String> kafkaOperator = env.addSource(flinkKafkaConsumer.setStartFromTimestamp(DateUtil.parse("2023-05-04 09:00", DatePattern.NORM_DATETIME_MINUTE_PATTERN).getTime()));
        //kafkaOperator.print();
        SingleOutputStreamOperator<List<CarTripOutputPhev>> process =kafkaOperator
                //Kafka message deserialize
                .map((MapFunction<String, CarTripPhev>) s -> JSONUtil.toBean(s, CarTripPhev.class))
                //.map((MapFunction<String, CarTripPhev>) s -> JSONUtil.toBean(s.substring(1,s.length()-1), CarTripPhev.class))
                //DataStream → KeyedStream By Vin
                .keyBy(CarTripPhev::getDeviceId)
                //对数据进行链式处理
                .map(new TripPhevMapFunction())
                .filter((carTripPhev -> ObjectUtil.isNotEmpty(carTripPhev)));
        SingleOutputStreamOperator<List<List<CarTripOutputPhev>>> processhbase =process.countWindowAll(1).process(new ProcessAllWindowFunction<List<CarTripOutputPhev>, List<List<CarTripOutputPhev>>, GlobalWindow>() {
            @Override
            public void process(ProcessAllWindowFunction<List<CarTripOutputPhev>, List<List<CarTripOutputPhev>>, GlobalWindow>.Context context, Iterable<List<CarTripOutputPhev>> iterable, Collector<List<List<CarTripOutputPhev>>> collector) throws Exception {
                List<List<CarTripOutputPhev>> list =  ListUtil.list(true, iterable.iterator());
                collector.collect(list);
            }
        });
        processhbase.addSink(new SinkHbasePhev()).name("HBase-Sink");
        // execute program
        env.execute("Flink Streaming Java API Skeleton");
    }

}